/*
    客户端模块：将客户端的业务处理模块和网络模块进行融合，提供四种客户端：
        服务注册客户端RegistryClient、服务发现客户端DiscoveryClient、rpc客户端RpcClient、topic客户端TopicClient
    1、服务注册客户端：将业务层Provider模块(rpc_registry.hpp)和网络层MuduoClient(net.hpp)模块融合，搭建一个服务注册客户端，用于向服务注册中心注册服务
    2、服务发现客户端：将业务层Discover模块(rpc_registry.hpp)和网络层MuduoClient(net.hpp)模块融合，搭建一个服务发现客户端，用于向服务注册中心发现服务
    3、rpc客户端：将业务层RpcCaller模块(rpc_client.hpp)、Discover模块(rpc_registry.hpp)和网路层MuduoClient(net.hpp)模块融合，搭建一个rpc调用客户端
        可以向提供服务的节点发起rpc调用，也能够支持向服务注册中心发现服务(由Discover模块来完成，发现服务这一过程不需要用户操作，根据构造函数的参数来选择是否开启，然后由客户端自主完成)
        使用者能够根据构造函数的参数决定是否开启Discover模块，如果开启，客户端先发现服务再发起请求，否则直接发起请求
        3.1、rpc客户端设计思想：
            如果我已经知道提供rpc服务节点的地址，直接使用ip&port(提供rpc服务节点的地址)创建对象发起请求即可
            如果不知道，则根据ip&port(服务注册中心地址)先去服务注册中心发现服务，然后选择一个服务节点发起请求
        3.2、问题：如果是第二种情况的话，我获取到了服务的节点，那么我应该采用长连接还是短连接：
            短连接：每次选择一个服务节点，实例化客户端对象发起rpc请求，调用完毕，关闭客户端
                优点：思想简单，不需要维护连接
                缺点：rpc调用效率稍低，每次调用都需要重新建立连接
                难点：如果采用异步调用，那么我发送完请求之后就不能立即销毁客户端，必须等待拿到响应之后才可以删除客户端
            长连接：使用一个连接池将服务节点地址和客户端维护起来，每次rpc调用的时候，如果已经有对应的客户端之间使用，没有的话就创建
                优点：进行二次相同的rpc调用，可以直接复用连接，效率高
                缺点：需要维护连接池
                难点：在服务下线之后，需要删除连接池中下线节点对应的客户端，而服务的下线不能直接在rpc客户端中感知到，需要通过业务层Discover模块中的连接断开回调来感知
                      所以就需要提供一个回调函数，注册给业务层Discover模块，当处理下线通知的时候调用
    4、topic客户端：将业务层TopicManager模块(rpc_topic.hpp)和网络层MuduoClient(net.hpp)模块融合，搭建一个主题发布与订阅客户端，可以向服务器发起主题创建/删除/订阅/取消订阅/发布消息的请求
*/
#pragma once
#include "rpc_caller.hpp"
#include "rpc_registry.hpp"
#include "requestor.hpp"
#include "rpc_topic.hpp"
#include "../common/dispatcher.hpp"

namespace ns_jsonrpc
{
    namespace ns_client
    {
        // 服务注册客户端
        class RegistryClient
        {
        public:
            using RegistryClientPtr = std::shared_ptr<RegistryClient>;

        public:
            RegistryClient(const std::string &ip, int port)
                : _requestor(std::make_shared<Requestor>()),
                  _provider(std::make_shared<Provider>(_requestor)),
                  _dispatcher(std::make_shared<Dispatcher>())
            {
                // 逐层设置回调
                // 1、业务处理层：将requestor中的onResponse注册给Dispatcher模块，当接收到RSP_SERVICE类型消息时，回调处理
                auto regCb = std::bind(&Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerCallback<BaseMessage>(MType::RSP_SERVICE, regCb);

                // 2、网络处理层：将Dispatcher中的onMessage注册给MuduoClient模块，当接收到消息时，回调处理
                auto dptCb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                _client = MuduoClientFactory::create(ip, port);
                _client->setMessageCallback(dptCb);
                _client->connect();
            }

            // 服务注册
            bool serviceRegistry(const std::string &method, const Address &host)
            {
                return _provider->serviceRegistry(_client->connection(), method, host);
            }

        private:
            Requestor::RequestorPtr _requestor;
            Provider::ProviderPtr _provider;
            Dispatcher::DispatcherPtr _dispatcher;
            BaseClient::BaseClientPtr _client;
        };

        // 服务发现客户端
        class DiscoveryClient
        {
        public:
            using DiscoveryClientPtr = std::shared_ptr<DiscoveryClient>;

        public:
            DiscoveryClient(const std::string &ip, int port, const Discoverer::OfflineCallback &offlineCallback)
                : _requestor(std::make_shared<Requestor>()),
                  _discoverer(std::make_shared<Discoverer>(_requestor, offlineCallback)),
                  _dispatcher(std::make_shared<Dispatcher>())
            {
                // 逐层设置回调
                // 1、业务处理层：将requestor中的onResponse注册给Dispatcher模块，当接收到RSP_SERVICE类型消息时，回调处理
                auto dsvCb = std::bind(&Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerCallback<BaseMessage>(MType::RSP_SERVICE, dsvCb);

                // 业务回调：服务上下通知回调
                auto ntfCb = std::bind(&Discoverer::onServiceRequest, _discoverer.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerCallback<ServiceRequest>(MType::REQ_SERVICE, ntfCb);

                // 2、网络处理层：将Dispatcher中的onMessage注册给MuduoClient模块，当接收到消息时，回调处理
                auto dptCb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                _client = MuduoClientFactory::create(ip, port);
                _client->setMessageCallback(dptCb);
                _client->connect();
            }

            // 服务发现
            bool serviceDiscovery(const std::string &method, Address &host)
            {
                return _discoverer->serviceDiscovery(_client->connection(), method, host);
            }

        private:
            Requestor::RequestorPtr _requestor;
            Discoverer::DiscovererPtr _discoverer;
            Dispatcher::DispatcherPtr _dispatcher;
            BaseClient::BaseClientPtr _client;
        };

        // rpc客户端（引入了服务发现客户端，处理逻辑是：可以根据构造参数决定是直接rpc调用，还是先服务发现再根据发现结果进行rpc调用）
        class RpcClient
        {
        public:
            using RpcCallerPtr = std::shared_ptr<RpcCaller>;

        public:
            // enableDiscovery表示是否启用服务发现，如果为true则ip&port表示服务注册中心地址，先发现再调用，如果为false则表示提供rpc服务的地址，直接进行rpc调用
            RpcClient(bool enableDiscovery, const std::string &ip, int port)
                : _enableDiscovery(enableDiscovery),
                  _requestor(std::make_shared<Requestor>()),
                  _caller(std::make_shared<RpcCaller>(_requestor)),
                  _dispatcher(std::make_shared<Dispatcher>())
            {
                // 设置回调
                auto rpcCb = std::bind(&Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerCallback<BaseMessage>(MType::RSP_RPC, rpcCb);

                if (_enableDiscovery)
                {
                    auto offlineCallback = std::bind(&RpcClient::delClient, this, std::placeholders::_1);
                    _discovery_client = std::make_shared<DiscoveryClient>(ip, port, offlineCallback);
                }
                else
                {
                    BaseClient::BaseClientPtr client = NewClient(Address(ip, port));
                }
            }

            // 同步
            bool call(const std::string &method, const Json::Value &params, Json::Value &result)
            {
                BaseClient::BaseClientPtr client = getClient(method);
                if (client.get() == nullptr)
                {
                    LOG_ERROR("failed to sync rpc call, no client for method: %s\n", method.c_str());
                    return false;
                }
                bool ret = _caller->call(client->connection(), method, params, result);
                if (ret == false)
                {
                    LOG_ERROR("failed to sync rpc call, call function error, method: %s\n", method.c_str());
                    return false;
                }
                return true;
            }

            // 异步
            bool call(const std::string &method, const Json::Value &params, RpcCaller::JsonAsyncResponse &result)
            {
                BaseClient::BaseClientPtr client = getClient(method);
                if (client.get() == nullptr)
                {
                    LOG_ERROR("failed to async rpc call, no client for method: %s\n", method.c_str());
                    return false;
                }
                bool ret = _caller->call(client->connection(), method, params, result);
                if (ret == false)
                {
                    LOG_ERROR("failed to async rpc call, call function error, method: %s\n", method.c_str());
                    return false;
                }
                return true;
            }

            // 回调
            bool call(const std::string &method, const Json::Value &params, const RpcCaller::JsonResponseCallBack &callback)
            {
                BaseClient::BaseClientPtr client = getClient(method);
                if (client.get() == nullptr)
                {
                    LOG_ERROR("failed to callback rpc call, no client for method: %s\n", method.c_str());
                    return false;
                }
                bool ret = _caller->call(client->connection(), method, params, callback);
                if (ret == false)
                {
                    LOG_ERROR("failed to callback rpc call, call function error, method: %s\n", method.c_str());
                    return false;
                }
                return true;
            }

        private:
            BaseClient::BaseClientPtr getClient(const std::string &method)
            {
                // 判断是哪种调用方式
                if (_enableDiscovery)
                {
                    // 服务发现
                    Address host;
                    if (!_discovery_client->serviceDiscovery(method, host))
                        return BaseClient::BaseClientPtr();
                    
                    // 先在连接池中查找是否存在连接，如果没有则创建
                    BaseClient::BaseClientPtr client = selClient(host);
                    if (client.get() != nullptr)
                        return client;
                    client = NewClient(host);
                    putClient(host, client);
                    return client;
                }
                else
                {
                    // 直接调用
                    return _connections.begin()->second;
                }
            }

            // 实例化一个客户端 & 设置回调 & 与服务节点建立连接
            BaseClient::BaseClientPtr NewClient(const Address &host)
            {
                BaseClient::BaseClientPtr client = MuduoClientFactory::create(host.first, host.second);
                auto dispatchMessageCallback = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                client->setMessageCallback(dispatchMessageCallback);
                client->connect();
                putClient(host, client);
                return client;
            }

        private:
            // 向连接池中插入一个连接
            void putClient(const Address &host, const BaseClient::BaseClientPtr &client)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _connections[host] = client;
            }

            // 从连接池中选择一个连接
            BaseClient::BaseClientPtr selClient(const Address &host)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _connections.find(host);
                if (it == _connections.end())
                    return BaseClient::BaseClientPtr();
                return it->second;
            }

            // 从连接池删除一个连接（这是一个回调函数，被注册给服务发现者，当服务下线的时候，自动清理连接池对象）
            void delClient(const Address &host)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _connections.erase(host);
            }

        public:
            // 仿函数，用于连接池中自定义key计算hash
            struct AddressHash
            {
                size_t operator()(const Address &host) const
                {
                    std::string key = host.first + std::to_string(host.second);
                    return std::hash<std::string>{}(key);
                }
            };

        private:
            bool _enableDiscovery;
            Requestor::RequestorPtr _requestor;
            RpcCaller::RpcCallerPtr _caller;
            Dispatcher::DispatcherPtr _dispatcher;
            DiscoveryClient::DiscoveryClientPtr _discovery_client;
            std::unordered_map<Address, BaseClient::BaseClientPtr, AddressHash> _connections; // 连接池
            std::mutex _mutex;
        };

        // topic客户端
        class TopicClient 
        {
        public:
            using TopicClientPtr = std::shared_ptr<TopicClient>;

        public:
            TopicClient(const std::string &ip, int port):
                    _requestor(std::make_shared<Requestor>()),
                    _topic_manager(std::make_shared<TopicManager>(_requestor)),
                    _dispatcher(std::make_shared<Dispatcher>()),
                    _client(MuduoClientFactory::create(ip, port))
            {
                // 设置回调
                // 1、主题请求的响应回调
                auto topicRspCb = std::bind(&Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerCallback<BaseMessage>(MType::RSP_TOPIC, topicRspCb);

                // 2、主题发布请求回调
                auto topicPublishReqCb = std::bind(&TopicManager::onTopicPublish, _topic_manager.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerCallback<TopicRequest>(MType::REQ_TOPIC, topicPublishReqCb);

                // 3、消息分发回调
                auto dptCb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                _client->setMessageCallback(dptCb);
                _client->connect();
            }

            // 主题创建
            bool topicCreate(const std::string &topicName)
            {
                return _topic_manager->topicCreate(topicName, _client->connection());
            }

            // 主题删除
            bool topicRemove(const std::string &topicName)
            {
                return _topic_manager->topicRemove(topicName, _client->connection());
            }

            // 主题订阅
            bool topicSubscribe(const std::string &topicName, const TopicManager::SubscribeCallback &subscribeCallback)
            {
                return _topic_manager->topicSubscribe(topicName, _client->connection(), subscribeCallback);
            }

            // 主题取消订阅
            bool topicCancel(const std::string &topicName)
            {
                return _topic_manager->topicCancel(topicName, _client->connection());
            }

            // 主题发布
            bool topicPublic(const std::string &topicName, const std::string &msg)
            {
                return _topic_manager->topicPublish(topicName, _client->connection(), msg);
            }

            void shutdown()
            {
                _client->shutdown();
            }

        private:
            Requestor::RequestorPtr _requestor;
            TopicManager::TopicManagerPtr _topic_manager;
            Dispatcher::DispatcherPtr _dispatcher;
            BaseClient::BaseClientPtr _client;
        };
    }
}